package biz.ws.ws.client;

import api.common.SpringContext;
import io.netty.bootstrap.AbstractBootstrap;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.CharsetUtil;
import io.netty.util.concurrent.Promise;
import lombok.Getter;
import org.springframework.context.ApplicationEvent;
import org.springframework.lang.NonNull;

import java.io.Closeable;
import java.io.IOException;
import java.net.URI;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

/**
 * @创建人 wsdc
 * @时间 2021/4/1
 * @描述 websocket 客户端
 */
public class WsClient implements Closeable {
    EventLoopGroup group;
    Bootstrap boot;

    OnDataReceive onDataReceive;

    boolean sendEvent = false;

    public WsClient() {
        this(false,OnDataReceive.empty);
    }

    public WsClient(boolean sendEvent, OnDataReceive onDataReceive) {
        group = new NioEventLoopGroup(1);
        boot = new Bootstrap();
        boot.option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS,5000)
                .option(ChannelOption.SO_REUSEADDR,true)
                .group(group)
                .handler(new LoggingHandler(LogLevel.INFO))
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        ChannelPipeline p = socketChannel.pipeline();
                        p.addLast(new ChannelHandler[]{new HttpClientCodec(),
                                new HttpObjectAggregator(1024 * 1024 * 10)});
                        p.addLast("hookedHandler", new WebSocketClientHandler());
                    }
                });

        this.onDataReceive = onDataReceive;
        this.sendEvent = sendEvent;
    }

    public void connect(String url, @NonNull Consumer<ChannelFuture> consumer) throws Exception{
        //URI websocketURI = new URI(String.format("ws://192.168.31.99:7777/ws/ksv.do?token=%d&loginFrom=11",i));

        URI websocketURI = new URI(url);
        HttpHeaders httpHeaders = new DefaultHttpHeaders();
        WebSocketClientHandshaker handshaker = WebSocketClientHandshakerFactory.newHandshaker(websocketURI, WebSocketVersion.V13, (String) null, true, httpHeaders);
        boot.connect(websocketURI.getHost(), websocketURI.getPort())
                .addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if(future.isSuccess()){
                            Channel channel = future.channel();
                            WebSocketClientHandler handler = (WebSocketClientHandler) channel.pipeline().get("hookedHandler");
                            handler.setHandshaker(handshaker);
                            handshaker.handshake(channel);

                            //  握手
                            handler.handshakeFuture().addListener(new ChannelFutureListener() {
                                @Override
                                public void operationComplete(ChannelFuture future) throws Exception {
                                    consumer.accept(future);
                                }
                            });
                        }else{
                            //  失败了也返回出去 外部做出判断
                            consumer.accept(future);
                        }

                    }
                });
    }

    @Override
    public void close() throws IOException {
        group.shutdownGracefully();
    }

    public class WebSocketClientHandler extends SimpleChannelInboundHandler<Object> {
        WebSocketClientHandshaker handshaker;
        ChannelPromise handshakeFuture;

        public void handlerAdded(ChannelHandlerContext ctx) {
            this.handshakeFuture = ctx.newPromise();
        }

        public WebSocketClientHandshaker getHandshaker() {
            return handshaker;
        }

        public void setHandshaker(WebSocketClientHandshaker handshaker) {
            this.handshaker = handshaker;
        }

        public ChannelPromise getHandshakeFuture() {
            return handshakeFuture;
        }

        public void setHandshakeFuture(ChannelPromise handshakeFuture) {
            this.handshakeFuture = handshakeFuture;
        }

        public ChannelFuture handshakeFuture() {
            return this.handshakeFuture;
        }

        protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
            //System.out.println("channelRead0  " + this.handshaker.isHandshakeComplete());
            Channel ch = ctx.channel();
            FullHttpResponse response;
            if (!this.handshaker.isHandshakeComplete()) {
                try {
                    response = (FullHttpResponse) msg;
                    //握手协议返回，设置结束握手
                    this.handshaker.finishHandshake(ch, response);
                    //设置成功
                    this.handshakeFuture.setSuccess();
                    //System.out.println("WebSocket Client connected! response headers[sec-websocket-extensions]:{}" + response.headers());
                } catch (WebSocketHandshakeException var7) {
                    FullHttpResponse res = (FullHttpResponse) msg;
                    String errorMsg = String.format("WebSocket Client failed to connect,status:%s,reason:%s", res.status(), res.content().toString(CharsetUtil.UTF_8));
                    this.handshakeFuture.setFailure(new Exception(errorMsg));
                }
            } else if (msg instanceof FullHttpResponse) {
                response = (FullHttpResponse) msg;
                //this.listener.onFail(response.status().code(), response.content().toString(CharsetUtil.UTF_8));
                throw new IllegalStateException("Unexpected FullHttpResponse (getStatus=" + response.status() + ", content=" + response.content().toString(CharsetUtil.UTF_8) + ')');
            } else {
                WebSocketFrame frame = (WebSocketFrame) msg;
                if(onDataReceive != null){
                    onDataReceive.receiveData(ctx,frame);
                }

                if(sendEvent){
                    SpringContext.context().publishEvent(new WebsocketDataEvent(frame,ctx));
                }
            }

            if(msg instanceof WebSocketFrame){

            }
        }
    }

    public static class WebsocketDataEvent extends ApplicationEvent {
        @Getter
        WebSocketFrame webSocketFrame;

        @Getter
        ChannelHandlerContext ctx;
        public WebsocketDataEvent(Object source,ChannelHandlerContext ctx) {
            super(source);
            webSocketFrame = (WebSocketFrame) source;
            this.ctx = ctx;
        }
    }

    public interface OnDataReceive{
        void receiveData(ChannelHandlerContext ctx,WebSocketFrame webSocketFrame);

        OnDataReceive empty = new OnDataReceive() {
            @Override
            public void receiveData(ChannelHandlerContext ctx, WebSocketFrame webSocketFrame) {

            }
        };
    }
}
